
/*-------------------------------------------------------------------------
 *
 * pg_largeobject.c
 *	  支持pg_largeobject关系操作的例程
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/catalog/pg_largeobject.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/genam.h"
#include "access/htup_details.h"
#include "access/sysattr.h"
#include "access/table.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/pg_largeobject.h"
#include "catalog/pg_largeobject_metadata.h"
#include "miscadmin.h"
#include "utils/acl.h"
#include "utils/fmgroids.h"
#include "utils/rel.h"


/*
 * 创建一个具有给定 LO 标识符的大对象。
 *
 * 我们通过在 pg_largeobject_metadata 中插入一条没有数据页的条目来创建一个新的大对象，
 * 这样该对象将显示为大小为 0。
 */
Oid LargeObjectCreate(Oid fc_loid)
{
	Relation	pg_lo_meta;
	HeapTuple	fc_ntup;
	Oid			fc_loid_new;
	Datum		fc_values[Natts_pg_largeobject_metadata];
	bool		fc_nulls[Natts_pg_largeobject_metadata];

	pg_lo_meta = table_open(LargeObjectMetadataRelationId,
							RowExclusiveLock);

	/*
	 * 插入大对象的元数据
	 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));

	if (OidIsValid(fc_loid))
		fc_loid_new = fc_loid;
	else
		fc_loid_new = GetNewOidWithIndex(pg_lo_meta,
									  LargeObjectMetadataOidIndexId,
									  Anum_pg_largeobject_metadata_oid);

	fc_values[Anum_pg_largeobject_metadata_oid - 1] = ObjectIdGetDatum(fc_loid_new);
	fc_values[Anum_pg_largeobject_metadata_lomowner - 1]
		= ObjectIdGetDatum(GetUserId());
	fc_nulls[Anum_pg_largeobject_metadata_lomacl - 1] = true;

	fc_ntup = heap_form_tuple(RelationGetDescr(pg_lo_meta),
						   fc_values, fc_nulls);

	CatalogTupleInsert(pg_lo_meta, fc_ntup);

	heap_freetuple(fc_ntup);

	table_close(pg_lo_meta, RowExclusiveLock);

	return fc_loid_new;
}

/*
 * 删除具有给定 LO 标识符的大对象。数据页和元数据都必须被删除。
 */
void LargeObjectDrop(Oid fc_loid)
{
	Relation	pg_lo_meta;
	Relation	pg_largeobject;
	ScanKeyData fc_skey[1];
	SysScanDesc fc_scan;
	HeapTuple	fc_tuple;

	pg_lo_meta = table_open(LargeObjectMetadataRelationId,
							RowExclusiveLock);

	pg_largeobject = table_open(LargeObjectRelationId,
								RowExclusiveLock);

	/*
	 * 从 pg_largeobject_metadata 中删除一个条目
	 */
	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_metadata_oid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_loid));

	fc_scan = systable_beginscan(pg_lo_meta,
							  LargeObjectMetadataOidIndexId, true,
							  NULL, 1, fc_skey);

	fc_tuple = systable_getnext(fc_scan);
	if (!HeapTupleIsValid(fc_tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("large object %u does not exist", fc_loid)));

	CatalogTupleDelete(pg_lo_meta, &fc_tuple->t_self);

	systable_endscan(fc_scan);

	/*
	 * 从 pg_largeobject 中删除所有关联条目
	 */
	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_loid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_loid));

	fc_scan = systable_beginscan(pg_largeobject,
							  LargeObjectLOidPNIndexId, true,
							  NULL, 1, fc_skey);
	while (HeapTupleIsValid(fc_tuple = systable_getnext(fc_scan)))
	{
		CatalogTupleDelete(pg_largeobject, &fc_tuple->t_self);
	}

	systable_endscan(fc_scan);

	table_close(pg_largeobject, RowExclusiveLock);

	table_close(pg_lo_meta, RowExclusiveLock);
}

/*
 * LargeObjectExists
 *
 * 我们不使用系统缓存来存储大对象元数据，以避免使用过多的本地内存。
 *
 * 此函数始终使用最新快照扫描系统目录，
 * 因此在大对象以只读模式打开时不应使用（因为以只读模式打开的大对象应视为相对于调用者的快照，而在读写模式下则相对于当前快照）。
 */
bool LargeObjectExists(Oid fc_loid)
{
	Relation	pg_lo_meta;
	ScanKeyData fc_skey[1];
	SysScanDesc fc_sd;
	HeapTuple	fc_tuple;
	bool		fc_retval = false;

	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_metadata_oid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_loid));

	pg_lo_meta = table_open(LargeObjectMetadataRelationId,
							AccessShareLock);

	fc_sd = systable_beginscan(pg_lo_meta,
							LargeObjectMetadataOidIndexId, true,
							NULL, 1, fc_skey);

	fc_tuple = systable_getnext(fc_sd);
	if (HeapTupleIsValid(fc_tuple))
		fc_retval = true;

	systable_endscan(fc_sd);

	table_close(pg_lo_meta, AccessShareLock);

	return fc_retval;
}
